package com.zhang.second.day04;

import com.zhang.second.utils.ClickEvent;
import com.zhang.second.utils.ClickSource;
import com.zhang.second.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 全窗口聚合函数和增量聚合函数结合
 *
 * @title: 每个url在每个5秒钟的滚动窗口中被浏览的次数
 * @author: zhang
 * @date: 2022/1/31 12:14
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new CountAgg(), new ResultWindow())
                .print();

        env.execute();
    }

    public static class ResultWindow extends ProcessWindowFunction<Long,UrlViewCountPerWindow,String,TimeWindow>{
        @Override
        public void process(String key, ProcessWindowFunction<Long, UrlViewCountPerWindow, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UrlViewCountPerWindow> out) throws Exception {
            //累加器只有一个值，就是增量聚合函数的统计结果
            out.collect(new UrlViewCountPerWindow(
                    key,
                    elements.iterator().next(),
                    context.window().getStart(),
                    context.window().getEnd()
            ));
        }
    }

    public static class CountAgg implements AggregateFunction<ClickEvent, Long, Long> {
        //创建空累加器
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        //定义输入数据和累加器的聚合规则，并返回新的累加器
        @Override
        public Long add(ClickEvent value, Long accumulator) {
            return accumulator + 1L;
        }

        //闭合窗口时，输出计算结果
        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return null;
        }
    }
}
